package day4

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{lit, window}
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.types.{IntegerType, StringType, StructType}

object SparkStreaming {
  def main(args: Array[String]): Unit = {
    //需要实现应用对于所收到数据的及时反馈
    //Spark Streaming专门针对流处理
    //整的Spark Streaming流处理
    System.setProperty("hadoop.home.dir", "D:\\hadoop")
    Logger.getLogger("org").setLevel(Level.OFF)

    val session = SparkSession.builder()
      .appName("SparkSQL-test")
      .master("local[*]")
      .getOrCreate()

    /*
      1.定义数据来源

      2.定义处理的方式

      3.处理的输出
     */

    //1.定义数据来源
      //需要虚拟机向端口输出数据，在LINUX当中安装一个软件包
//    val line = session.readStream.format("socket")
//      .option("host","192.168.35.131")
//      .option("port","9999")
//      .load()

    //定义数据来源
    //1.定义文件的格式
    val fileSchema = new StructType()
      .add("key",StringType)
      .add("value",StringType)


    val line = session.readStream.format("text")
      .schema(fileSchema)
      .load("D:\\data\\HCIP\\files")

    //2.定义数据处理
    import session.implicits._
    val get = line.select("value").as[String].flatMap(_.split("\t"))
//      .map((_,1))
//      .groupByKey(_._1).count()



    //val get = line.select("value").as[String].flatMap(_.split(" "))

    //定义输出
    val put = get.writeStream
      .format("console")
      .outputMode("update")
      .trigger(Trigger.ProcessingTime("1 second"))
    //.trigger(Trigger.ProcessingTime("5 seconds"))


    //4 执行
    put.start().awaitTermination()


    /**
     * outputMode有三种选项：
     * complete: 每次输出都会将所有数据完整的输出；只支持输出内容比输入内容少，才能够实现将数据保存在内存中
     * append：只会输出新内容，以前内容不会再输出
     * update：只会输出上次处理结果变更之后的数据，绝大部分的操作都支持
     */

    /**
     * trigger：触发器
     * 固定内容：ProcessingTime指定间隔时间
     * Once：使用的是Once触发器，在接收到数据之后只处理一次，可以在外部再次定义Once触发器，实现触发自定义
     * Continuous：使用Continuous触发器连续触发
     */



  }

}
